<?php
/**
 * Copyright (c) [2019] [吴跃忠]
 * [selibra] is licensed under the Mulan PSL v1.
 * You can use this software according to the terms and conditions of the Mulan PSL v1.
 * You may obtain a copy of Mulan PSL v1 at:
 * http://license.coscl.org.cn/MulanPSL
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
 * PURPOSE.
 * See the Mulan PSL v1 for more details.
 */

namespace LeoMQ\Realizes\RabbitMQ;


use LeoMQ\Intf\ConsumerInterface;
use LeoMQ\Intf\MQSubscriber;
use LeoMQ\MessageHandler;
use LeoMQ\Annotations\ConsumerRealize;
use Selibra\Tools\Console;
use Swoole\Process;

#[ConsumerRealize("rabbitmq")]
class Consumer implements ConsumerInterface
{

    /**
     * @param MQSubscriber $subscriber
     * @throws \ErrorException
     */
    public function listen(MQSubscriber $subscriber)
    {
        $process = new Process(function () use ($subscriber) {
            go(function () use ($subscriber) {
                $connection = (new Connection())->getConnection();
                $channel = $connection->channel();
                $channel->queue_declare($subscriber->queue(), false, false, false, false);

                $callback = function ($msg) use ($subscriber) {
                    // 获取消费者
                    $handler = new MessageHandler($msg->body, $subscriber);
                    try {
                        $handler->handle();
                    } catch (\Exception $exception) {
                        return false;
                    }
                    // 消息响应
                    return $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
                };
                $channel->basic_consume($subscriber->queue(), $subscriber->consumerTag(), false, false, false, false, $callback);

                while (count($channel->callbacks)) {
                    $channel->wait();
                }
            });
        });
        $process->start();
    }

}